package com.kristin.zklock;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class ZkLock implements Lock {

    private static final String LOCK_NAME = "lock_stock";

    private static final String ROOT_PATH = "/lock";

    private ZooKeeper zk;

    ThreadLocal<String> currNode = new ThreadLocal<String>();


    {
        try {
            zk = new ZooKeeper("127.0.0.1:2181", 20 * 1000, null);
            // 判断root节点是否存在
            if (zk.exists(ROOT_PATH, null) == null) {
                // 创建一个持久化节点
                zk.create(ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

    }

    static class NodeWatch implements Watcher {

        private CountDownLatch countDownLatch;

        public NodeWatch(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        public void process(WatchedEvent event) {
            Event.EventType type = event.getType();
            switch (type) {
                case NodeDeleted:
                    System.out.println("节点已删除");
                    countDownLatch.countDown();
                    break;
            }
        }
    }


    public void lock() {
        // 去获取锁，查看自己是否所序列中最小的
        try {
            // 创建一个临时序列节点

            String myNode = zk.create(ROOT_PATH + "/" + LOCK_NAME, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 创建临时节点
            System.out.println("创建临时节点：" + myNode);
            List<String> children = zk.getChildren(ROOT_PATH, null);
            // 排序
            TreeSet<String> treeSet = new TreeSet<String>();
            for (String node : children) {
                treeSet.add(ROOT_PATH + "/" + node);
            }

            String firstNode = treeSet.first();

            if (firstNode.equals(myNode)) {
                // 第一个节点就是自己，表示自己获取到锁
                currNode.set(myNode);
                return;
            }

            // 监听前一个节点
            String preNodeId = treeSet.lower(myNode);

            // 注册监听
            System.out.println("注册监听");
            CountDownLatch countDownLatch = new CountDownLatch(1);
            zk.getData(preNodeId, new NodeWatch(countDownLatch), null);

            countDownLatch.await();
            currNode.set(myNode);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void lockInterruptibly() throws InterruptedException {

    }

    public boolean tryLock() {
        return false;
    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    public void unlock() {
        // 删除临时节点
        try {
            System.out.println("删除临时节点:" + currNode.get());
            zk.delete(currNode.get(), -1);
            currNode.remove();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }
}
